package com.comtop.topcloud.device.service.kafka;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.comtop.topcloud.device.mapper.EnergyEfficiencyAssessmentMapper;
import com.comtop.topcloud.device.mapper.EnergyMonitorAlarmMapper;
import com.comtop.topcloud.device.model.AlarmConfig;
import com.comtop.topcloud.device.model.Command;
import com.comtop.topcloud.device.model.EnergyConfig;
import com.comtop.topcloud.device.model.influxdb.DeviceReportingDataModel;
import com.comtop.topcloud.device.model.influxdb.GatewayUsageModel;
import com.comtop.topcloud.device.model.mysql.EnergyEfficiencyAssessment;
import com.comtop.topcloud.device.model.mysql.EnergyEquipmentMonitorRefTag;
import com.comtop.topcloud.device.model.mysql.EnergyMonitorAlarm;
import com.comtop.topcloud.device.model.mysql.EventTable;
import com.comtop.topcloud.device.service.DeviceBaseInfoService;
import com.comtop.topcloud.device.service.EnergyEfficiencyManagementService;
import com.comtop.topcloud.device.service.EnergyEquipmentMonitorRefTagService;
import com.comtop.topcloud.device.service.EnergySystemAlarmLibraryService;
import com.comtop.topcloud.device.service.EventTableService;
import com.comtop.topcloud.device.util.ResultMapper;
import com.comtop.topcloud.device.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.influxdb.annotation.Column;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import plus.ojbk.influxdb.core.InfluxdbTemplate;

import javax.annotation.Resource;
import java.lang.reflect.Field;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Service
@Slf4j

public class KafkaConsumerService {

    @Resource
    private DeviceBaseInfoService deviceBaseInfoService;
    @Resource
    private InfluxdbTemplate influxdbTemplate;
    @Resource
    private EventTableService eventTableService;
    @Resource
    private EnergyEquipmentMonitorRefTagService energyEquipmentMonitorRefTagService;
    @Resource
    private EnergySystemAlarmLibraryService energySystemAlarmLibraryService;
    @Resource
    private EnergyEfficiencyManagementService energyEfficiencyManagementService;
    @Resource
    private EnergyMonitorAlarmMapper energyMonitorAlarmMapper;
    @Resource
    private EnergyEfficiencyAssessmentMapper energyEfficiencyAssessmentMapper;

    public static final ThreadPoolExecutor TX = new ThreadPoolExecutor(10, 10, 60, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000));

    @KafkaListener(topics = "energy_topic_01", groupId = "my-group", containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listenEnergyTopic01(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        JSONObject jsonObject = null;
        try {
            // 解析 JSON 对象
            jsonObject = JSON.parseObject(record.value());
            log.info("Received record: {}", jsonObject);

            if (jsonObject == null) {
                log.warn("Received record is null. Skipping processing.");
                acknowledgment.acknowledge();
                return;
            }

            JSONObject notifyData = jsonObject.getJSONObject("notify_data");
            if (notifyData == null) {
                log.warn("notify_data is null in jsonObject: {}", jsonObject);
                acknowledgment.acknowledge();
                return;
            }

            JSONObject body = notifyData.getJSONObject("body");
            if (body == null) {
                log.warn("Body is null in notify_data: {}", notifyData);
                acknowledgment.acknowledge();
                return;
            }

            String topic = body.getString("topic");
            JSONObject content = body.getJSONObject("content");

            if (StringUtils.isBlank(topic) || content == null) {
                log.warn("Content or topic is null. Content: {}, Topic: {}", content, topic);
                acknowledgment.acknowledge();
                return;
            }

            // 定义正则模式
            String topicPatternInfo = "/edge/.+/.+/info";
            String topicPatternStatus = "/edge/.+/.+/status";
            String topicPatternRtg = "/edge/.+/.+/rtg";
            String topicPatternCmdSetCack = "/edge/.+/.+/cmd/set/cack";

            // 使用正则匹配 topic，调用相应方法
            if (Pattern.matches(topicPatternInfo, topic)) {
                log.info("Matching topic pattern: /edge/.+/.+/info");
                listenInfoTopic(content);
            } else if (Pattern.matches(topicPatternStatus, topic)) {
                log.info("Matching topic pattern: /edge/.+/.+/status");
                listenStatusTopic(content);
            } else if (Pattern.matches(topicPatternRtg, topic)) {
                log.info("Matching topic pattern: /edge/.+/.+/rtg");
                listenRtgTopic(content);
            } else if (Pattern.matches(topicPatternCmdSetCack, topic)) {
                log.info("Matching topic pattern: /edge/.+/.+/cmd/set/cack");
                listenSetCackTopic(content);
            } else {
                log.warn("Topic does not match any predefined patterns: {}", topic);
            }

            // 手动提交偏移量
            acknowledgment.acknowledge();
            log.info("Message processed and offset acknowledged.");

        } catch (Exception e) {
            // 记录处理失败的 JSON 对象，帮助调试
            log.error("Error processing message, failed JSON object: {}", jsonObject, e);
        }
    }


    //@KafkaListener(topicPattern = "/edge/.+/.+/info", groupId = "my-group",containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listenInfoTopic(JSONObject content) {

        try {
            // 解析上报的设备信息
            GatewayUsageModel gatewayUsageModel = ResultMapper.mapJsonToGatewayUsageModel(content);

            // 查询设备参数信息，并填充到上报的设备信息中
            String monitorName = gatewayUsageModel.getMonitorName();
            String serialNumber = gatewayUsageModel.getSn();
            GatewayUsageModel usageModel = deviceBaseInfoService.getGatewayUsageModelByDeviceNameAndSerialNumber(monitorName, serialNumber);
            //List<DeviceParameterValue> deviceParameterValueList = deviceBaseInfoService.getDeviceParameterValueByDeviceName(deviceName);
            //ResultMapper.mapDeviceParametersToObject(deviceParameterValueList,usageModel);
            BeanUtil.copyProperties(usageModel, gatewayUsageModel, new CopyOptions().ignoreNullValue());

            if (!validateTagFields(gatewayUsageModel)) {
                throw new RuntimeException("上报数据中缺少标签字段");
            }
            // 写入InfluxDB
            influxdbTemplate.insert(gatewayUsageModel);

        } catch (Exception e) {
            throw e;
        }
    }

    /**
     * 监听 /edge/${pKey}/${sn}/status
     */
    //@KafkaListener(topicPattern = "/edge/.+/.+/status", groupId = "my-group",containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listenStatusTopic(JSONObject content) {
        try {
            // 解析上报的设备工况
            GatewayUsageModel gatewayUsageModel = ResultMapper.mapJsonToGatewayUsageModel(content);

            // 查询设备参数信息，并填充到上报的设备信息中
            String monitorName = gatewayUsageModel.getMonitorName();
            String serialNumber = gatewayUsageModel.getSn();
            GatewayUsageModel usageModel = deviceBaseInfoService.getGatewayUsageModelByDeviceNameAndSerialNumber(monitorName, serialNumber);
            BeanUtil.copyProperties(usageModel, gatewayUsageModel, new CopyOptions().ignoreNullValue());

            if (!validateTagFields(gatewayUsageModel)) {
                throw new RuntimeException("上报数据中缺少标签字段");
            }
            // 写入InfluxDB
            influxdbTemplate.insert(gatewayUsageModel);

        } catch (Exception e) {
            throw e;
        }
    }

    /**
     * 监听 /edge/${pKey}/${sn}/rtg
     */
    //@KafkaListener(topicPattern = "/edge/.+/.+/rtg", groupId = "my-group",containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listenRtgTopic(JSONObject content) {
        try {
            // 解析物联网关上报实时数据
            List<DeviceReportingDataModel> deviceReportingDataModels = ResultMapper.mapJsonToDeviceReportingDataModels(content);
            Iterator<DeviceReportingDataModel> iterator = deviceReportingDataModels.iterator();
            // 查询设备参数信息，并填充到上报的设备信息中
            while (iterator.hasNext()) {
                DeviceReportingDataModel deviceReportingDataModel = iterator.next();
                String monitorName = deviceReportingDataModel.getMonitorName();
                String serialNumber = deviceReportingDataModel.getSerialNumber();
                EnergyEquipmentMonitorRefTag energyEquipmentMonitorRefTag = energyEquipmentMonitorRefTagService.getEquipmentMonitorRefTag(serialNumber, monitorName, deviceReportingDataModel.getTagName());
                if (energyEquipmentMonitorRefTag == null) {
                    log.warn("Unable to find EnergyEquipmentMonitorRef for serialNumber: {}, deviceName: {}, tagName: {}",
                            serialNumber, monitorName, deviceReportingDataModel.getTagName());
                    iterator.remove();
                    continue;
                }
                if (StringUtils.isNotEmpty(energyEquipmentMonitorRefTag.getLon()) || StringUtils.isNotEmpty(energyEquipmentMonitorRefTag.getLat())) {
                    deviceReportingDataModel.setLocation(energyEquipmentMonitorRefTag.getLon() + "," + energyEquipmentMonitorRefTag.getLat());
                }
                BeanUtil.copyProperties(energyEquipmentMonitorRefTag, deviceReportingDataModel);
                deviceReportingDataModel.setDeviceId(energyEquipmentMonitorRefTag.getEquipmentId());
                if (!validateTagFields(deviceReportingDataModel)) {
                    log.warn("上报数据中缺少标签字段,:{}", deviceReportingDataModel);
                    iterator.remove();
                    continue;
                }
                TX.execute(() -> dataCheck(energyEquipmentMonitorRefTag, deviceReportingDataModel));
                TX.execute(() -> energyManagerCheck(energyEquipmentMonitorRefTag, deviceReportingDataModel));
            }

            //log.info("数据:{}",deviceReportingDataModels);
            // 写入InfluxDB
            influxdbTemplate.insert(deviceReportingDataModels);

        } catch (Exception e) {
            throw e;
        }
    }

    private void energyManagerCheck(EnergyEquipmentMonitorRefTag refTag, DeviceReportingDataModel dataModel) {
        List<EnergyConfig> energyConfigs = energyEfficiencyManagementService.getEnergyConfigByMonitorType(refTag.getMonitorType());
        if (CollUtil.isEmpty(energyConfigs)) {
            return;
        }
        String localDateTime = TimeUtils.formatByLocalDateTimeFormat(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss");
        List<EnergyEfficiencyAssessment> assessmentList = energyConfigs.parallelStream()
                .filter(energyConfig -> !energyConfig.isNull())
                .flatMap(energyConfig ->
                        CollUtil.emptyIfNull(energyConfig.getProcessingPlans()).stream()
                                .filter(processingPlan -> handleCondition(processingPlan.getOperation(), Double.valueOf(processingPlan.getIndicatorValue()), dataModel.getVal()))
                                .map(processingPlan -> {
                                    EnergyEfficiencyAssessment assessment = new EnergyEfficiencyAssessment();
                                    assessment.setId(UUID.randomUUID().toString().replace("-", StringUtils.EMPTY).toUpperCase());
                                    assessment.setProjectName(refTag.getProjectName());
                                    assessment.setProjectType(energyConfig.getProjectType());
                                    assessment.setIndicatorName(energyConfig.getIndicatorName());
                                    assessment.setCurrentValue(String.valueOf(dataModel.getVal()));
                                    assessment.setStandardValue(energyConfig.getStandardValue());
                                    assessment.setPerfectValue(energyConfig.getPerfectValue());
                                    assessment.setCurrentStatusAssessment(processingPlan.getHandlePlan());
                                    assessment.setCreateTime(localDateTime);
                                    return assessment;
                                })
                ).collect(Collectors.toList());
        if (CollUtil.isNotEmpty(assessmentList)) {
            energyEfficiencyAssessmentMapper.batchInsert(assessmentList);
        }
    }


    public void dataCheck(EnergyEquipmentMonitorRefTag energyEquipmentMonitorRefTag, DeviceReportingDataModel reportingDataModel) {
        List<AlarmConfig> list = energySystemAlarmLibraryService.getAlarmConfigByDeviceTypeAndTerminalType(energyEquipmentMonitorRefTag.getEquipmentType(), energyEquipmentMonitorRefTag.getMonitorType());
        Timestamp currentTimestamp = Timestamp.valueOf(LocalDateTime.now());
        List<EnergyMonitorAlarm> energyMonitorAlarmList = list.parallelStream()
                .filter(alarmConfig -> !alarmConfig.isNull() &&
                        compareFields(alarmConfig.getTerminalFields(), energyEquipmentMonitorRefTag.getMonitorName()) &&
                        compareFields(alarmConfig.getTagFields(), energyEquipmentMonitorRefTag.getTagName()))
                .flatMap(alarmConfig ->
                        alarmConfig.getEnergySystemAlarmConditions().stream()
                                .filter(c -> compareValues(c.getCondition1(), Double.valueOf(c.getAlarmValue1()), reportingDataModel.getVal()))
                                .map(alarmCondition -> {
                                    // 将时间戳获取移到外部，避免每个元素都调用一次
                                    EnergyMonitorAlarm energyMonitorAlarm = new EnergyMonitorAlarm();
                                    energyMonitorAlarm.setAlarmName(alarmConfig.getAlarmName());
                                    energyMonitorAlarm.setAlarmLevel(alarmCondition.getAlarmLevel());
                                    energyMonitorAlarm.setAlarmTime(currentTimestamp);  // 使用外部的时间戳
                                    energyMonitorAlarm.setAlarmStatus(0);
                                    energyMonitorAlarm.setTerminalFields(energyEquipmentMonitorRefTag.getMonitorName());
                                    energyMonitorAlarm.setEquipmentType(energyEquipmentMonitorRefTag.getEquipmentType());
                                    energyMonitorAlarm.setOrgId(energyEquipmentMonitorRefTag.getOrgId());
                                    energyMonitorAlarm.setValue(String.valueOf(reportingDataModel.getVal()));
                                    energyMonitorAlarm.setAlarmDescribe(alarmCondition.getRemark1());
                                    energyMonitorAlarm.setEquipmentId(energyEquipmentMonitorRefTag.getEquipmentId());
                                    energyMonitorAlarm.setId(UUID.randomUUID().toString());
                                    energyMonitorAlarm.setAlarmType(alarmConfig.getAlarmType());
                                    energyMonitorAlarm.setTagName(reportingDataModel.getTagName());
                                    if (StringUtils.isNotEmpty(energyEquipmentMonitorRefTag.getTagNameAliasName())) {
                                        energyMonitorAlarm.setTagNameAliasName(energyEquipmentMonitorRefTag.getTagNameAliasName());
                                    }
                                    return energyMonitorAlarm;
                                })
                ).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(energyMonitorAlarmList)) {
            energyMonitorAlarmMapper.batchInsert(energyMonitorAlarmList);
        }
    }

    public boolean compareFields(String field, String name) {
        if (StringUtils.isEmpty(field)) {
            return true;
        }
        String[] fields = StringUtils.split(field, ",");

        // 将 deviceName 转换为大写
        String nameUpper = name.toUpperCase();

        // 遍历分割后的数组，比较每个元素与 deviceName 大写后的值
        for (String f : fields) {
            if (f != null && f.toUpperCase().equals(nameUpper)) {
                return true;  // 如果找到一致的项，返回 true
            }
        }

        return false;  // 如果没有找到一致的项，返回 false
    }


    public boolean compareValues(String condition, Double reportingValue, Double alarmValue1) {
        if (condition == null) {
            return false; // 如果条件为空，返回 false
        }

        // 定义一个误差范围，用于比较浮点数
        final double epsilon = 1e-6;  // 你可以根据实际需求调整误差范围

        switch (condition) {
            case ">":
                return reportingValue > alarmValue1;
            case ">=":
                return reportingValue >= alarmValue1;
            case "<":
                return reportingValue < alarmValue1;
            case "<=":
                return reportingValue <= alarmValue1;
            case "=":
                // 使用 Math.abs 比较两个 Double 值是否足够接近
                return Math.abs(reportingValue - alarmValue1) < epsilon;
            default:
                return false; // 如果条件没有匹配的符号，返回 false
        }
    }


    private boolean handleCondition(String operation, Double indicatorValue, Double deviceValue) {
        // 如果条件为空，返回 false
        if (operation == null) {
            return false;
        }
        switch (operation) {
            case ">":
                return deviceValue > indicatorValue;
            case ">=":
                return deviceValue >= indicatorValue;
            case "<":
                return deviceValue < indicatorValue;
            case "<=":
                return deviceValue <= indicatorValue;
            default:
                return false; // 如果条件没有匹配的符号，返回 false
        }
    }

    /**
     * 监听 /edge/${pKey}/${sn}/cmd/set/cack
     */
    //@KafkaListener(topicPattern = "/edge/.+/.+/cmd/set/cack", groupId = "my-group",containerFactory = "kafkaManualAckListenerContainerFactory")
    public void listenSetCackTopic(JSONObject content) {
        try {
            // 解析响应控制命令
            Command command = ResultMapper.mapJsonToCommand(content);

            // 检查 command 或 command.data 是否为空
            if (BeanUtil.isEmpty(command)) {
                log.warn("解析 command 失败，content: {}", content);
                return;
            }

            if (BeanUtil.isEmpty(command.getData())) {
                log.warn("command 中的 data 为空，content: {}", content);
                return;
            }

            // 查询事件表中的设备信息
            EventTable eventTable = eventTableService.getById(command.getData().getSysid());
            if (eventTable == null) {
                log.warn("未找到对应的事件记录，sysid: {}", command.getData().getSysid());
                return;
            }

            // 更新事件表中的相关信息
            eventTable.setRemarks(command.getRemark());
            eventTable.setStatus(command.getValid() == 1 ? "Success" : "Failed");
            eventTable.setResponseData(JSONObject.toJSONString(content));
            eventTable.setResponseTimestamp(command.getTs());

            // 更新事件记录到数据库
            eventTableService.updateById(eventTable);
            log.info("成功更新事件记录，sysid: {}", command.getData().getSysid());

        } catch (Exception e) {
            // 捕获异常并记录详细信息，抛出自定义异常或者重新抛出异常
            log.error("处理响应控制命令时发生异常，content: {}", content, e);
            // 根据需要选择抛出自定义异常或直接抛出
            throw new RuntimeException("处理响应控制命令时失败", e);
        }
    }

    public <T> boolean validateTagFields(T model) {
        if (model == null) {
            return false;  // 如果 model 为 null，直接返回 false
        }

        // 获取所有字段
        Field[] fields = model.getClass().getDeclaredFields();

        for (Field field : fields) {
            // 检查字段是否有 tag = true 的注解
            if (field.isAnnotationPresent(Column.class)) {
                Column columnAnnotation = field.getAnnotation(Column.class);

                // 如果该字段是 tag 字段且其值为 null，则返回 false
                if (columnAnnotation.tag()) {
                    try {
                        // 如果字段是私有的，需要设置可访问
                        field.setAccessible(true);

                        // 如果该字段的值为 null，返回 false
                        if (field.get(model) == null) {
                            log.warn("字段 {} 的值为 null，验证失败", field.getName());
                            return false;
                        }
                    } catch (IllegalAccessException e) {
                        log.error("访问字段 {} 时出现异常：{}", field.getName(), e.getMessage());
                        // 如果发生异常，直接返回 false
                        return false;
                    }
                }
            }
        }

        // 如果所有 tag 字段都不为 null，返回 true
        return true;
    }


}
